package com.credithc.signature.callback.mq.core;

import com.alibaba.fastjson.JSONObject;
import com.credithc.eagleeye.plugins.thread.EagleEyeRunnable;
import com.credithc.signature.callback.service.MessageService;
import com.credithc.signature.callback.utils.ProtobufUtil;
import com.credithc.signature.common.ro.callback.CallbackReq;
import com.credithc.signature.common.utils.MessageFilterUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class CoreCallbackConsumer implements ChannelAwareMessageListener {

    @Autowired
    MessageService messageService;

    private static final String MDC_REQUEST_SN = "mdc_request_sn";

    ThreadPoolExecutor threadPoolExecutor;

    public CoreCallbackConsumer() {
        this.threadPoolExecutor = new ThreadPoolExecutor(10, 50, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(200), new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        CallbackReq callbackReq = null;
        try {
//            callbackReq = ProtobufUtil.deserializer(message.getBody(), CallbackReq.class);
            callbackReq = JSONObject.parseObject(message.getBody(), CallbackReq.class);
            log.info("获取到回调消息体:callbackReq={}", callbackReq);
            errorMessageFilter(callbackReq);
            this.threadPoolExecutor.execute(new EagleEyeRunnable(new CoreCallbackTask(callbackReq)));
//            messageService.process(callbackReq);
        } catch (Throwable e) {
            if (callbackReq != null) {
                log.error("消息处理失败......", e);
            } else {
                log.error("消息处理失败发回mq", e);
            }
        } finally {
            try {
                // 确认消息成功接收
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Throwable e) {
                // 不做处理
            }
            try {
                MDC.remove(MDC_REQUEST_SN);
            } catch (Exception e) {
                // 不做处理
            }
        }
    }

    /**
     * 过滤错误信息中的特殊字符--防止页面查询出错
     *
     * @param callbackReq
     */
    private void errorMessageFilter(CallbackReq callbackReq) {
        //过滤特殊字符
        String errorMsg = callbackReq.getErrorMsg();
        errorMsg = MessageFilterUtil.filter(errorMsg);
        callbackReq.setErrorMsg(errorMsg);
        //限制异常错误信息长度
        errorMsg = MessageFilterUtil.limitLength(errorMsg);
        callbackReq.setErrorMsg(errorMsg);
    }

    /**
     * 多线程任务队列
     */
    class CoreCallbackTask implements Runnable, Serializable {

        CallbackReq callbackReq;

        public CoreCallbackTask(CallbackReq callbackReq) {
            this.callbackReq = callbackReq;
        }

        @Override
        public void run() {
            messageService.process(callbackReq);
        }
    }

}
